package com.hy;


import com.rabbitmq.client.Channel;
import convert.ImageMessageConverter;
import convert.PDFMessageConverter;
import convert.TextMessageConverter;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * Created by huangyu on 2018/9/23 20:24
 */
@Configuration
@ComponentScan({"com.hy.*"})
public class RabbitMqConfig {

	@Bean
	public ConnectionFactory connectionFactory() {
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
		connectionFactory.setAddresses("192.168.177.129:5672");
		connectionFactory.setUsername("guest");
		connectionFactory.setPassword("guest");
		connectionFactory.setVirtualHost("/");
		return connectionFactory;
	}

	@Bean
	public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
		rabbitAdmin.setAutoStartup(true);
		return rabbitAdmin;

	}

	@Bean
	public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
		return rabbitTemplate;

	}

	/**
	 * 消息消费者监听容器
	 *
	 * @param connectionFactory
	 * @return
	 */
	@Bean
	public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
		SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
		simpleMessageListenerContainer.setQueues(queue001(), queue002(), queue003(), queue_image(), queue_pdf());
		simpleMessageListenerContainer.setConcurrentConsumers(1);
		simpleMessageListenerContainer.setMaxConcurrentConsumers(5);
		simpleMessageListenerContainer.setDefaultRequeueRejected(false);
		simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
		simpleMessageListenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {
			@Override
			public String createConsumerTag(String queueName) {
				return queueName + "_" + UUID.randomUUID().toString();
			}
		});

		/**
		 * 下面演示两种方式 监听消息 获得消费的消息
		 * 1.简单消息监听容器
		 * 2.消息监听适配器
		 */

		/*简单消息监听容器*/
//		simpleMessageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
//			@Override
//			public void onMessage(Message message, Channel channel) throws Exception {
//				String msg = new String(message.getBody());
//				System.out.println(msg);
//			}
//		});

		/*  消息监听适配器
		    默认调用的方法是handleMessage
			可以自己设置为consumeMessage
			 Converter把字节数组消息转String
		*/
//		MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//		adapter.setDefaultListenerMethod("consumeMessage");
//		adapter.setMessageConverter(new TextMessageConverter());
//		simpleMessageListenerContainer.setMessageListener(adapter);
//		return simpleMessageListenerContainer;

		/*
			可以指定: 队列名称和方法名绑定
		*/
//		MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//		adapter.setDefaultListenerMethod("consumeMessage");
//		adapter.setMessageConverter(new TextMessageConverter());
//		Map<String, String> queueOrTagToMethodName = new HashMap<>();
//		queueOrTagToMethodName.put("queue001", "method1");
//		queueOrTagToMethodName.put("queue002", "method2");
//		adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
//		simpleMessageListenerContainer.setMessageListener(adapter);
//		return simpleMessageListenerContainer;


		/**
		 * 下面是MessageConvert消息转换器讲解
		 */

		// 1.1 支持json格式的转换器
		/**MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
		adapter.setDefaultListenerMethod("consumeMessage");

		Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
		adapter.setMessageConverter(jackson2JsonMessageConverter);

		simpleMessageListenerContainer.setMessageListener(adapter);*/


		// 1.2 DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持java对象转换
		/**
		 MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
		 adapter.setDefaultListenerMethod("consumeMessage");

		 Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();

		 DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
		 jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);

		 adapter.setMessageConverter(jackson2JsonMessageConverter);
		 container.setMessageListener(adapter);
		 */


		//1.3 DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持java对象多映射转换
		/**
		 MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
		 adapter.setDefaultListenerMethod("consumeMessage");
		 Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
		 DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();

		 Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();
		 idClassMapping.put("order", com.bfxy.spring.entity.Order.class);
		 idClassMapping.put("packaged", com.bfxy.spring.entity.Packaged.class);

		 javaTypeMapper.setIdClassMapping(idClassMapping);

		 jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);
		 adapter.setMessageConverter(jackson2JsonMessageConverter);
		 container.setMessageListener(adapter);
		 */

		//1.4 ext convert

		MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
		adapter.setDefaultListenerMethod("consumeMessage");

		//全局的转换器:
		ContentTypeDelegatingMessageConverter convert = new ContentTypeDelegatingMessageConverter();

		TextMessageConverter textConvert = new TextMessageConverter();
		convert.addDelegate("text", textConvert);
		convert.addDelegate("html/text", textConvert);
		convert.addDelegate("xml/text", textConvert);
		convert.addDelegate("text/plain", textConvert);

		Jackson2JsonMessageConverter jsonConvert = new Jackson2JsonMessageConverter();
		convert.addDelegate("json", jsonConvert);
		convert.addDelegate("application/json", jsonConvert);

		ImageMessageConverter imageConverter = new ImageMessageConverter();
		convert.addDelegate("image/png", imageConverter);
		convert.addDelegate("image", imageConverter);

		PDFMessageConverter pdfConverter = new PDFMessageConverter();
		convert.addDelegate("application/pdf", pdfConverter);


		adapter.setMessageConverter(convert);
		simpleMessageListenerContainer.setMessageListener(adapter);

		return simpleMessageListenerContainer;
	}

	/**
	 * [针对消费者配置]
	 * 1.设置交换机类型
	 * 2.将队列绑定到交换机
	 * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
	 * HeadersExchange ：通过添加属性key-value匹配
	 * DirectExchange:按照routingkey分发到指定队列
	 * TopicExchange:多关键字匹配
	 */
	@Bean
	public TopicExchange exchange001() {
		return new TopicExchange("topic001", true, false);
	}

	@Bean
	public Queue queue001() {
		return new Queue("queue001", true);
	}

	@Bean
	public Binding binding001() {
		return BindingBuilder.bind(queue001()).to(exchange001()).with("spring.*");
	}

	@Bean
	public TopicExchange exchange002() {
		return new TopicExchange("topic002", true, false);
	}

	@Bean
	public Queue queue002() {
		return new Queue("queue002", true);
	}

	@Bean
	public Binding binding002() {
		return BindingBuilder.bind(queue002()).to(exchange002()).with("rabbit.*");
	}

	@Bean
	public Queue queue003() {
		return new Queue("queue003", true);
	}

	@Bean
	public Binding binding003() {
		return BindingBuilder.bind(queue003()).to(exchange001()).with("mq.*");
	}

	@Bean
	public Queue queue_image() {
		return new Queue("image_queue", true);
	}

	@Bean
	public Queue queue_pdf() {
		return new Queue("pdf_queue", true);
	}
}
